package com.deepglint.window;

import com.deepglint.beans.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.*;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @author mj
 * @version 1.0
 * @date 2021-11-17 22:12
 */
public class WindowTest_TimeWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String path = "C:\\Users\\马军\\Desktop\\Idea-workspace\\flink\\src\\main\\resources\\source.txt";
        DataStreamSource<String> streamSource = env.readTextFile("path");

        SingleOutputStreamOperator<SensorReading> streamMap = streamSource.map((MapFunction<String, SensorReading>) value -> {
            String[] split = value.split(" ");
            return new SensorReading(split[0], split[1], new Long(split[2]), new Double(split[3]));
        });

        // 开窗操作.分配bucket
        // 如果直接windowAll。则会失去并行度
        // 如果不想失去并行度，需要keyby后，进行开窗操作
        streamMap
//                .windowAll(SlidingEventTimeWindows.of(Time.seconds(15), Time.seconds(5)))
                .keyBy("id")
//                .countWindow(100) // 滚动计数窗口(窗口时间为计数为100个，计数100个后，再开一个窗口)
//                .countWindow(100,5) // 滑动计数窗口(窗口时间为计数为100个，计数5个后，再开一个窗口)
//                .window(EventTimeSessionWindows.withGap(Time.seconds(15))) // 会话时间窗口(窗口时间为15秒，过后不开窗口)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(15))) // 滚动时间窗口(窗口长度15秒,过15秒后，再开一个窗口)
//                .window(SlidingProcessingTimeWindows.of(Time.seconds(15), Time.seconds(5))) // 滑动时间窗口(窗口长度15秒，过5秒后再次开一个窗口)
        ;

        env.execute();
    }
}
